package com.atguigu.gmall.realtime.dim.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.bean.TableProcessDim;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.HBaseUtil;
import com.atguigu.gmall.realtime.common.util.RedisUtil;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

/**
 * ClassName: HbaseSinkFunction
 * Description:
 * 将流中数据同步到Hbase
 *
 * @Create 2024/4/17 14:23
 */
public class HbaseSinkFunction extends RichSinkFunction<Tuple2<JSONObject, TableProcessDim>> {

    Connection hBaseConnection;
    Jedis jedis;

    @Override
    public void open(Configuration parameters) throws Exception {
        hBaseConnection = HBaseUtil.getHBaseConnection();
        jedis = RedisUtil.getJedis();
    }

    @Override
    public void close() throws Exception {
        HBaseUtil.closeHBaseConn(hBaseConnection);
        RedisUtil.closeJedis(jedis);
    }

    //将流中的数据同步到Hbase对应的表中
    @Override
    public void invoke(Tuple2<JSONObject, TableProcessDim> tup2, Context context) throws Exception {
        //维度数据
        JSONObject dataJsonObject = tup2.f0;

        //获取当前维度数据对应的配置信息
        TableProcessDim tableProcessDim = tup2.f1;

        //获取业务数据库对维度表的配置信息
        String type = dataJsonObject.getString("type");

        //type没必要留了
        dataJsonObject.remove("type");

        String sinkTable = tableProcessDim.getSinkTable();
        //从dataJsonObject中取出id的值
        String rowKey = dataJsonObject.getString(tableProcessDim.getSinkRowKey());

        if ("delete".equals(type)) {
            //从Hbase中删除对应的维度数据
            HBaseUtil.delRow(hBaseConnection, Constant.HBASE_NAMESPACE, sinkTable, rowKey);
        } else {
            //将维度数据put到Hbase表中
            String sinkFamily = tableProcessDim.getSinkFamily();
            HBaseUtil.putRow(hBaseConnection, Constant.HBASE_NAMESPACE, sinkTable, rowKey, sinkFamily, dataJsonObject);
        }

        //TODO （DWS旁路缓存）如果业务数据库维度表发生了变化，将Redis缓存的数据清除掉
        if ("delete".equals(type) || "update".equals(type)) {
            jedis.del(sinkTable + ":" + rowKey);
        }
    }
}
